package com.zf.rabbitmq.consumer.listener.consume_confirm;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * 消费确认
 */
@Component
public class ConsumeConfirmConsumer {

    /**
     * ackMode = "MANUAL" 表示手动确认消息，默认情况下消息是自动确认消费的。
     * <p>
     * 注意：
     * 1、如果一个deliveryTag已经被消费或者拒绝了，那么对于当前通达就不会再出现同样的 deliveryTag，当再次确认或者拒绝该 deliveryTag 时（也就是确认或者拒绝一个不存的消息），
     * rabbitmq会返回一个找不到deliveryTag的错误信息，本地客户端会关闭当前通道，当前通道无法再继续消费，也就时该通代表的消费者关闭了，不再消费消息。
     * <p>
     * 2、但是如果集成了SpringBoot来执行 1 中的步骤则会在通道关闭后创建一个新的的通道来进行消费，然后因为 1 中的步骤 再次重复 “关闭通道->创建新通道“
     */
    @RabbitListener(queues = {"test_consume_confirm_queue"}, ackMode = "MANUAL")
    public void consumeConfirm(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        // 消息标识
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        // 是否是重新入队的消息
        Boolean redelivered = message.getMessageProperties().getRedelivered();

        if (redelivered) {
            // 如果是重新入队消息则直接丢弃
            channel.basicNack(deliveryTag, false, false);
            System.out.printf("deliveryTag：%d，消息：[ %s ] 是重新入队消息，直接丢弃，不再入队\n", deliveryTag, msg);
            // 直接返回，不再往下执行，防止确认或者拒绝一个不存在的消息，造成当前通道关闭，当前通道无法再消费消息，同时抛出一个通道关闭的异常。
            // 但是SpringBoot会再次打开一个新的通道来消费消息，然后再次重复上述过程，会不停的关闭通道和创建新的通道。
            // 但是如果没有集成SpringBoot，而是使用原始的amqp-client则关闭了就是关闭了也不会再创建新的的通道。
            // 所以最好加上return，这样通道不会关闭，可以一直使用同一个通道
            // 这里使用return实现不再往下执行，也可以把下面的语句放到else中
            return;
        }

        if (deliveryTag % 2 == 0) {
            // 确认消费消息
            channel.basicAck(deliveryTag, false);
            System.out.printf("deliveryTag：%d，消息：[ %s ] 被消费\n", deliveryTag, msg);
        } else {
            // 拒绝消费消息，并重新入队
            channel.basicReject(deliveryTag, true);
            System.out.printf("deliveryTag：%d，消息：[ %s ] 被拒绝了且重新入队\n", deliveryTag, msg);
        }
    }
}
